﻿/*
   Copyright 2012 Dmitry Bratus

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
*/
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Flower.Client;
using Flower.Services.Data;
using Flower.Workflow;
using Flower.Directory.Util;
using Flower.Processing.Interruptions;
using Flower.Services;

namespace Flower.Processing.Activities
{
    [SaveAfter]
    [Transactional]
    [Retry]
    public class Enqueue<TMsg> : WaitBase
    {
        private interface IEnqueue
        {
            bool Enqueue(Message message);
        }

        private class SetEnqueue : IEnqueue
        {
            private readonly Processor _processor;
            private readonly string _subset;

            public SetEnqueue(Processor processor, string subset)
            {
                _processor = processor;
                _subset = subset;
            }

            public bool Enqueue(Message message)
            {
                _processor.Log.Trace(() => string.Format("Executing __enqueueMessage({0}, <msg>, {1}).", _subset, _processor.Pid));

                return ScriptHelper.Enqueue(_processor.Client, _subset, message, _processor.Pid);
            }
        }

        private class QueueEnqueue : IEnqueue
        {
            private readonly Processor _processor;
            private readonly IQueueService _queue;
            private readonly string _recipient;
            private readonly string _servicePath;

            public QueueEnqueue(Processor processor, IQueueService queue, string recipient, string servicePath)
            {
                _processor = processor;
                _queue = queue;
                _recipient = recipient;
                _servicePath = servicePath;
            }

            public bool Enqueue(Message message)
            {
                _processor.Log.Trace(() => string.Format("Executing IQueueService.Enqueue({0}, <msg>) at {1}.", _recipient, _servicePath));

                _queue.Enqueue(_recipient, message.Data);
                return true;
            }
        }

        internal Func<IContext, string> Subset { get; set; }
        internal Func<IContext, string> MessageName { get; set; }
        internal Func<IContext, TMsg> Message { get; set; }
        internal Func<IContext, MessageFormat> Format { get; set; }

        internal Enqueue() { }

        public override bool IsBlocking
        {
            get { return true; }
        }

        private static IEnqueue ResolvePath(Processor processor, string path)
        {
            if (path == null)
            {
                var servicePath = string.Concat("/Processors/", processor.ProcessorName, "/queue");
                var queue = processor.Client.GetService<IQueueService>(servicePath);

                return new QueueEnqueue(processor, queue, string.Concat(processor.Pid, "/", typeof(TMsg).FullName), processor.Pid);
            }
            else if (DirectoryUtil.IsId(path))
            {
                IQueueService queue = processor.Client.GetProcessQueue(path);

                return new QueueEnqueue(processor, queue, string.Concat(path, "/", typeof(TMsg).FullName), path);
            }
            else if (path.StartsWith("/Services"))
            {
                string servicePath, recipient;
                DirectoryUtil.CutPathTailSegment(path, out servicePath, out recipient);

                var queue = processor.Client.GetService<IQueueService>(servicePath);

                return new QueueEnqueue(processor, queue, recipient, servicePath);
            }
            else
            {
                string subset = DirectoryUtil.LocalSubsetToGlobal(processor.Pid, path);

                return new SetEnqueue(processor, subset);
            }
        }

        public override Interruption Do(Processor processor)
        {
            string subset = (Subset != null) ? Subset(processor) : null;

            IEnqueue enq = ResolvePath(processor, subset);

            TMsg payload = Message(processor);

            var message = new Message
            {
                Name = (MessageName != null) ? MessageName(processor) : null,
                Data = (payload != null) ? payload.ToBlob(Convert(Format(processor))) : null
            };

            if (!enq.Enqueue(message))
            {
                if (Interrupt)
                {
                    return Waiting.Instance;
                }
                else
                {
                    if (enq is SetEnqueue)
                    {
                        processor.State.__WaitLinks.Add
                        (
                            string.Concat(DirectoryUtil.LocalSubsetToGlobal(processor.Pid, subset), "/PutWaiters/", processor.Pid)
                        );
                    }

                    ++processor.State.__ActivityPointer;
                    return null;
                }
            }
            else
            {
                if (WaitingFinished != null)
                {
                    WaitingFinished(processor);
                }

                processor.State.__ActivityPointer = Position;
                return null;
            }
        }
    }
}
